package SparkSql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import java.util.List;

/**
 * 使用反射的方式将RDD转换为DataFrame
 */
public class RDD2DataFrameReflection {
    public static void main(String[] args) {
        //创建普通的RDD
        SparkConf conf = new SparkConf().setMaster("local").setAppName("RDD2DataFrameReflection");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);

        JavaRDD<String> lines = sc.textFile("");
        JavaRDD<Student> students = lines.map(
                (Function<String, Student>) line -> {
                    String[] lineSpilted = line.split(",");
                    Student stu = new Student();
                    stu.setId(Integer.valueOf(lineSpilted[0].trim()));
                    stu.setName(lineSpilted[1]);
                    stu.setAge(Integer.valueOf(lineSpilted[2].trim()));
                    return stu;
                }
        );

        //使用反射方式，将RDD转换为DataFrame
        //将Student.class传入进去，其实就是用反射的方式来创建DataFrame
        //因为Student.class本身就是反射的一个应用
        //然后底层还得通过对Student.class进行反射，来获取其中的field
        //JavaBean，必须实现Serializable接口，是可序列化的
        Dataset<Row> studentDF = sqlContext.createDataFrame(students, Student.class);

        //拿到了一个DataFrame后，将其注册为一个临时表，然后针对其中的数据执行SQL语句
        studentDF.registerTempTable("students");

        //针对students临时表执行SQL语句，查询年龄小于18岁的学生
        Dataset<Row> teenagerDF = sqlContext.sql("select * from student where age <= 18");
        //将查询出来的DataFrame。再次转换为RDD
        JavaRDD<Row> teenagerRDD = teenagerDF.javaRDD();

        //将RDD中的数据，进行映射，映射为student
        JavaRDD<Student> teenagerStudentRDD = teenagerRDD.map(
                (Function<Row, Student>) row -> {
                    //row中的数据，可能与原数据顺序不一致
                    Student student = new Student();
                    student.setId(row.getInt(0));
                    student.setName(row.getString(1));
                    student.setId(row.getInt(2));
                    return student;
                }
        );

        //将数据collect回来，打印出来
        List<Student> studentList = teenagerStudentRDD.collect();
        for (Student stu : studentList) {
            System.out.println(stu);
        }

    }
}
